package com.weimin.springbootgrpcclient.service.impl;

import com.weimin.proto.BaseGrpcServiceGrpc;
import com.weimin.proto.BaseMessage;
import com.weimin.springbootgrpcclient.service.IGrpcService;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class GrpcServiceImpl implements IGrpcService {

    @GrpcClient("grpc-server-name")
    BaseGrpcServiceGrpc.BaseGrpcServiceBlockingStub blockingStub;

    @GrpcClient("grpc-server-name")
    BaseGrpcServiceGrpc.BaseGrpcServiceStub serviceStub;


    @Override
    public BaseMessage.Message commonMethod(String msg) {
        BaseMessage.MessageRequest request = BaseMessage.MessageRequest.newBuilder()
                .setId(String.valueOf(System.currentTimeMillis()))
                .setData("commonMethod 请求数据 -> " + msg).build();
        BaseMessage.Message message = BaseMessage.Message.newBuilder()
                .setPackType(BaseMessage.Message.PackType.REQUEST)
                .setRequest(request).build();
        return blockingStub.commonMethod(message);
    }

    @Override
    public List<BaseMessage.Message> serviceEndStreamMethod(String msg) {
        BaseMessage.MessageRequest request = BaseMessage.MessageRequest.newBuilder()
                .setId(String.valueOf(System.currentTimeMillis()))
                .setData("serviceEndStreamMethod 请求数据 -> " + msg).build();
        BaseMessage.Message message = BaseMessage.Message.newBuilder()
                .setPackType(BaseMessage.Message.PackType.REQUEST)
                .setRequest(request).build();
        List<BaseMessage.Message> list = new ArrayList<>();
        Iterator<BaseMessage.Message> messageIterator = blockingStub.serviceEndStreamMethod(message);
        while (messageIterator.hasNext()) {
            BaseMessage.Message response = messageIterator.next();
            BaseMessage.MessageResponse messageRequest = response.getResponse();
            String id = messageRequest.getCode();
            String data = messageRequest.getData();
            log.info("clientStreamMethod 回复->类型 {} ID {} 回复数据 {}", response.getPackType(), id, data);
            list.add(response);
        }
        return list;
    }

    @Override
    public BaseMessage.Message clientStreamMethod(List<BaseMessage.Message> list) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // 继承StreamObserver接口
        ResponseStreamObserver<BaseMessage.Message> streamObserver = new ResponseStreamObserver<BaseMessage.Message>() {

            private BaseMessage.Message message;

            @Override
            public BaseMessage.Message getResult() {
                return this.message;
            }

            @Override
            public void onNext(BaseMessage.Message message) {
                BaseMessage.MessageResponse messageRequest = message.getResponse();
                String id = messageRequest.getCode();
                String data = messageRequest.getData();
                log.info("clientStreamMethod 回复->类型 {} ID {} 回复数据 {}", message.getPackType(), id, data);
                this.message = message;
            }

            @Override
            public void onError(Throwable throwable) {
                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                countDownLatch.countDown();
            }
        };
        StreamObserver<BaseMessage.Message> messageStreamObserver = serviceStub.clientStreamMethod(streamObserver);
        // 往服务端发送数据
        for (BaseMessage.Message message : list) {
            messageStreamObserver.onNext(message);
        }
        messageStreamObserver.onCompleted();
        // 设置等待超时时间60秒
        countDownLatch.await(60, TimeUnit.SECONDS);
        return streamObserver.getResult();
    }

    @Override
    public List<BaseMessage.Message> bidirectionalFlowMethod(List<BaseMessage.Message> list) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // 继承StreamObserver接口
        ResponseStreamObserver<BaseMessage.Message> streamObserver = new ResponseStreamObserver<BaseMessage.Message>() {

            private final List<BaseMessage.Message> list = new ArrayList<>();

            @Override
            public List<BaseMessage.Message> getResultList() {
                return this.list;
            }

            @Override
            public void onNext(BaseMessage.Message message) {
                BaseMessage.MessageResponse messageRequest = message.getResponse();
                String id = messageRequest.getCode();
                String data = messageRequest.getData();
                log.info("bidirectionalFlowMethod 回复->类型 {} ID {} 回复数据 {}", message.getPackType(), id, data);
                this.list.add(message);
            }

            @Override
            public void onError(Throwable throwable) {
                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                countDownLatch.countDown();
            }
        };
        StreamObserver<BaseMessage.Message> messageStreamObserver = serviceStub.bidirectionalFlowMethod(streamObserver);
        // 往服务端发送数据
        for (BaseMessage.Message message : list) {
            messageStreamObserver.onNext(message);
        }
        messageStreamObserver.onCompleted();
        // 设置等待超时时间60秒
        countDownLatch.await(60, TimeUnit.SECONDS);
        return streamObserver.getResultList();
    }
}
